package cn.com.sparkle.paxos.net.client.system;

import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.config.Configuration;
import cn.com.sparkle.paxos.deamon.ReConnectDeamon;
import cn.com.sparkle.paxos.deamon.ReConnectDeamon.ReConnectMethod;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.event.events.NodeStateChangeEvent;
import cn.com.sparkle.paxos.net.client.NetNode;
import cn.com.sparkle.paxos.net.frame.FrameBody;
import cn.com.sparkle.paxos.net.netlayer.NetClient;
import cn.com.sparkle.paxos.net.netlayer.NetHandler;
import cn.com.sparkle.paxos.net.netlayer.PaxosSession;
import cn.com.sparkle.paxos.net.netlayer.PaxosSessionKeys;
import cn.com.sparkle.paxos.net.protocolprocessor.DefaultProtocolProcessorChain;
import cn.com.sparkle.paxos.net.protocolprocessor.ProtocolManager;
import cn.com.sparkle.paxos.net.protocolprocessor.ProtocolProcessorChain;
import cn.com.sparkle.paxos.net.protocolprocessor.filter.ConnectCheckFilter;

public class SystemClientHandler implements NetHandler {
	private final NetClient netClient;
	private ReentrantLock lock = new ReentrantLock();
	private EventsManager eventsManager;
	private ReConnectDeamon reConnectThread;
	private Configuration conf;

	private ProtocolProcessorChain<FrameBody> processor;

	private SystemClientNegotiationProcessor negotiationProcessor;

	public SystemClientHandler(NetClient netClient, EventsManager eventsManager, ReConnectDeamon reConnectThread, Configuration conf,
			ProtocolManager protocolManager) {
		this.netClient = netClient;
		this.eventsManager = eventsManager;
		this.reConnectThread = reConnectThread;
		this.conf = conf;

		negotiationProcessor = new SystemClientNegotiationProcessor(conf, protocolManager, eventsManager);
		processor = new DefaultProtocolProcessorChain<FrameBody>();
		processor.addFirst(negotiationProcessor);
		processor.addFirst(ConnectCheckFilter.getInstance());
	}

	@Override
	public void onDisconnect(PaxosSession session) {
		NetNode nnode = session.get(PaxosSessionKeys.NET_NODE_KEY);
		if (nnode != null) {
			nnode.close();
			try {
				lock.lock();
				NodeStateChangeEvent.doLoseConnectEvent(eventsManager, nnode);
			} finally {
				lock.unlock();
			}
		}
		String address = session.get(PaxosSessionKeys.ADDRESS_KEY);
		reConnect(address);
		processor.onDisConnect(session);
	}

	@Override
	public void onConnect(PaxosSession session, Object attachment) {
		session.put(PaxosSessionKeys.ADDRESS_KEY, attachment);
		processor.onConnect(session);
		negotiationProcessor.negotiation(session, conf.getSelfAddress());
	}

	@Override
	public void onRecieve(PaxosSession session, FrameBody body) {
		processor.receive(body, session);
	}

	@Override
	public void onRefuse(Object connectAttachment) {
		String address = (String) connectAttachment;
		reConnect(address);
	}

	private final class ReConnect implements ReConnectMethod {
		private final Logger log = Logger.getLogger(ReConnect.class);

		@Override
		public void reConnect(Object value) {
			String address = (String) value;
			String[] a = address.split(":");
			try {
				if (conf.isDebugLog()) {
					log.debug("reConnect " + address);
				}
				netClient.connect(a[0], Integer.parseInt(a[1]), address);
			} catch (Throwable e) {
//				reConnectThread.add(address, this, 2000);
				SystemClientHandler.this.reConnect(address);
			}
		}

	}

	private final ReConnect method = new ReConnect();

	private void reConnect(String address) {
		// After sleep 5 seconds,retry to connect.
		
		if(address == null){
			System.err.println(address);
		}
		reConnectThread.add(address, method, 2000);
	}

}
